package com.zuikaku.source;

import com.zuikaku.pojo.Order;
import com.zuikaku.util.TimeUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.math.BigDecimal;
import java.util.*;

/**
 * 自定义Source，OrderSource，目标输出Order对象
 */
public class OrderSource extends RichParallelSourceFunction<Order> {
    private volatile boolean canceled = false;

    private List<String> itemNameList = Arrays.asList("iPhone12","红米K70","联想拯救者","华硕飞行堡垒");
    //固定价格，方便观察
    private List<BigDecimal> priceList = Arrays.asList(new BigDecimal(10),new BigDecimal(5),new BigDecimal(15),new BigDecimal(12));
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        System.out.println("OrderSource 已开启");
    }

    @Override
    public void close() throws Exception {
        super.close();
        System.out.println("OrderSource 已关闭");
    }

    /**
     * 执行产出order的逻辑，此处为mock的数据
     * Tips:run只执行一次，只不过内部实在持续的拿数据
     * @param sourceContext
     * @throws Exception
     */
    @Override
    public void run(SourceContext<Order> sourceContext) throws Exception {
        //mock 拿出数据
        while(!canceled){
            Thread.sleep(1000);//每1s产生一个随机订单
            Order order = new Order();
            Random random =new Random();
            int itemIdx = random.nextInt(itemNameList.size());
            order.setOrderId(UUID.randomUUID().toString());
            order.setPrice(priceList.get(itemIdx));
            order.setCustomerId(1000+random.nextInt(9999));
            order.setItemName(itemNameList.get(itemIdx));
            order.setCreateDate(new Date());
            System.out.println("产生："+order.getItemName()+" 价格："+order.getPrice()+" 时间："+ TimeUtil.formatTime(order.getCreateDate()));
            sourceContext.collect(order);
        }
    }

    @Override
    public void cancel() {
        canceled = true;
        System.out.println("OrderSource已取消");
    }
}
